package com.warren.financial.lease.realtime.app.func;


import com.alibaba.fastjson.JSONObject;
import com.warren.financial.lease.realtime.bean.TableProcess;
import com.warren.financial.lease.realtime.common.FinancialLeaseCommon;
import com.warren.financial.lease.realtime.util.HBaseUtil;
import com.warren.financial.lease.realtime.util.RedisUtil;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

/**
 * @AUTHOR warren
 * @date 2023/12/13
 **/
public class DimSinkFunc extends RichSinkFunction<Tuple3<String, JSONObject, TableProcess>> {

    private Connection hBaseConnection;

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        hBaseConnection = HBaseUtil.getHBaseConnection();
        jedis = RedisUtil.getRedisClient();
    }

    @Override
    public void invoke(Tuple3<String, JSONObject, TableProcess> value, Context context) throws Exception {
        // 拆分value
        String type = value.f0;
        JSONObject data = value.f1;
        TableProcess tableProcess = value.f2;

        // 写到hbase的表格
        String sinkTable = tableProcess.getSinkTable();

        // 写倒hbase的主键
        String sinkRowKey = tableProcess.getSinkRowKey();
        String rowKeyValue = data.getString(sinkRowKey);
        String sinkFamily = tableProcess.getSinkFamily();

        String[] columns = tableProcess.getSinkColumns().split(",");
        String[] values = new String[columns.length];
        for (int i = 0; i < values.length; i++) {
            values[i] = data.getString(columns[i]);
        }

        if("delete".equals(type)){
            HBaseUtil.deleteRow(hBaseConnection, FinancialLeaseCommon.HBASE_NAMESPACE,sinkTable,rowKeyValue);
        }else{
            // 写入倒对应的维度数据
            HBaseUtil.putRow(hBaseConnection, FinancialLeaseCommon.HBASE_NAMESPACE, sinkTable,rowKeyValue,sinkFamily,columns,values);
        }

        //同步删除 Redis 中的维度数据
        if("delete".equals(type) || "update".equals(type)){
            jedis.del(sinkTable + ":" + rowKeyValue);
        }
    }

    @Override
    public void close() throws Exception {
        HBaseUtil.closeHBaseConnection(hBaseConnection);
    }
}
